/**
 * Copyright (C), 2020-2020, 软件公司
 * FileName: RPCClient
 * Author:   cakin
 * Date:     2020/5/12
 * Description: 消费端以及测试类
 */
package com.rabbitmq.rpc;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.util.ConnectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.UUID;

import java.util.concurrent.TimeoutException;

/**
 * @ClassName: RPCClient
 * @Description: 消费端以及测试类
 * @Date: 2020/5/12
 * @Author: cakin
 */
@SuppressWarnings("deprecation")
@Slf4j
public class RPCClient {
    /**
     * 连接
     */
    private Connection connection;
    /**
     * 信道
     */
    private Channel channel;
    /**
     * 请求队列
     */
    private String requestQueueName = "rpc_queue";
    /**
     * 响应队列
     */
    private String replyQueueName;
    /**
     * 消费者
     */
    private QueueingConsumer consumer;

    public RPCClient() throws IOException, TimeoutException {
        // 创建消息和信道
        connection = ConnectionUtils.getConnection();
        channel = connection.createChannel();
        // 响应队列是匿名队列
        replyQueueName = channel.queueDeclare().getQueue();
        // 创建一个消费者
        consumer = new QueueingConsumer(channel);
        // 消费消息，推模式，监听
        channel.basicConsume(replyQueueName, true, consumer);
    }

    public String call( String message ) throws IOException,
            ShutdownSignalException, ConsumerCancelledException,
            InterruptedException {
        String response = null;
        // 使用UUID创建CorrelationId
        String corrId = UUID.randomUUID().toString();
        // props 携带CorrelationId 和 replyQueueName
        BasicProperties props = new BasicProperties.Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();
        // 发送消息到服务端，这里绑定的是默认exchange
        channel.basicPublish("", requestQueueName, props, message.getBytes());

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            // 接收客户端发过来的消息，接收前判断CorrelationId是否一致
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response = new String(delivery.getBody());
                break;
            }
        }

        return response;
    }

    public void close() throws Exception {
        connection.close();
    }

    public static void main( String[] args ) throws Exception {
        // 创建客户端
        RPCClient fibRpc = new RPCClient();
        log.info(" [x] Requesting fib(30)");
        // 请求服务端计算斐波那契数值
        String response = fibRpc.call("30");
        log.info(" [.] Got '" + response + "'");
        fibRpc.close();
    }
}


